package rpc.netty.http;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.multipart.Attribute;
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
import io.netty.handler.codec.http.multipart.InterfaceHttpData;
import io.netty.util.CharsetUtil;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadFactory;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;
import org.springframework.util.CollectionUtils;
import rpc.bootstrap.spring.SpringBootstrap;
import rpc.core.FlowReq;
import rpc.core.FlowRes;
import rpc.core.InitClose;
import rpc.core.plugin.HttpPrePlugin;
import rpc.core.plugin.HttpRetPlugin;
import rpc.trace.Trace;
import rpc.trace.TraceContext;
import rpc.utils.JsonUtil;
import rpc.utils.TypeSafe;


/*********
 *
 *
 *@author 197
 *@date 2020/7/16 18:50
 *
 **/

@Sharable
public class HttpServer extends SimpleChannelInboundHandler<FullHttpRequest> implements InitClose {

  private static final Logger logger = LoggerFactory.getLogger(HttpServer.class);
  private final int maxPackageSize = 1000_000_000;
  private int coreThreads = 20;
  private int port;
  private boolean nativeNetty = false;
  private ServerBootstrap serverBootstrap;
  private EventLoopGroup bossGroup;
  private EventLoopGroup workerGroup;
  private Channel serverChannel;
  private int workerThreads = 20;
  private Environment environment;

  private List<HttpPrePlugin> httpPrePluginList;
  private List<HttpRetPlugin> httpRetPluginList;

  public HttpServer setEnvironment(Environment environment) {
    this.environment = environment;
    return this;
  }

  public HttpServer setHttpPrePluginList(List<HttpPrePlugin> httpPrePluginList) {
    this.httpPrePluginList = httpPrePluginList;
    return this;
  }

  public HttpServer setHttpRetPluginList(List<HttpRetPlugin> httpRetPluginList) {
    this.httpRetPluginList = httpRetPluginList;
    return this;
  }

  public HttpServer setPort(int port) {
    this.port = port;
    return this;
  }

  @Override
  public void init() {

    coreThreads = TypeSafe.anyToInt(environment.getProperty("rpc.http.boss.threads"), 8);
    workerThreads = TypeSafe.anyToInt(environment.getProperty("rpc.http.worker.threads"), 8);
    String osName = System.getProperty("os.name");
    String linux = "linux";
    if (nativeNetty && StringUtils.isNotBlank(osName) && osName.toLowerCase().contains(linux)) {
      nativeNetty = true;
    }
    ThreadFactory bossThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("rpc_nettyHttp_boss_%d").build();
    ThreadFactory workThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("rpc_nettyHttp_worker_%d").build();
    if (nativeNetty) {
      bossGroup = new EpollEventLoopGroup(coreThreads, bossThreadFactory);
      workerGroup = new EpollEventLoopGroup(workerThreads, workThreadFactory);
    } else {
      bossGroup = new NioEventLoopGroup(coreThreads, bossThreadFactory);
      workerGroup = new NioEventLoopGroup(workerThreads, workThreadFactory);
    }
    serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);

    serverBootstrap.option(ChannelOption.SO_BACKLOG, 128);
    serverBootstrap.option(ChannelOption.SO_REUSEADDR, true);
    serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
    serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
    serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        //以下代码在 HttpClientPipelineFactory.java getPipeline()中
        ChannelPipeline pipeline = ch.pipeline()
            .addLast("decoder", new HttpRequestDecoder())
            .addLast("encoder", new HttpResponseEncoder())
            .addLast("aggregator", new HttpObjectAggregator(512 * 1024))
            .addLast(HttpServer.this);
      }
    });
  }

  @Override
  public void start() {
    InetSocketAddress addr = new InetSocketAddress(this.port);
    serverChannel = serverBootstrap.bind(addr).syncUninterruptibly().channel();
    logger.info("netty server started on host(*) port(" + port + ")");
  }

  @Override
  public void close() {

  }

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
    // 获取请求的uri
    int endIndex = req.uri().indexOf("?");
    String uri = req.uri();
    if (endIndex > 0) {
      uri = uri.substring(0, endIndex);
    }

    HttpMethod method = req.method();
    final Map<String, Object> parmMap = new HashMap<>(3);

    if (HttpMethod.GET == method) {
      // 是GET请求
      QueryStringDecoder decoder = new QueryStringDecoder(req.uri());
      // entry.getValue()是一个List, 只取第一个元素
      decoder.parameters().forEach((k, v) -> parmMap.put(k, v.get(0)));
    } else if (HttpMethod.POST == method) {
      // 是POST请求
      if ((req.headers().get("Content-Type") + "").contains("application/json")) {
        String params = req.content().toString(Charset.defaultCharset());
        parmMap.putAll(JsonUtil.fromJson(params));
      } else {
        HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(req);
        decoder.offer(req);
        List<InterfaceHttpData> paramList = decoder.getBodyHttpDatas();
        for (InterfaceHttpData param : paramList) {
          Attribute data = (Attribute) param;
          parmMap.put(data.getName(), data.getValue());
        }
      }
    }

    FlowRes flowRes = new FlowRes();
    int retCode = 0;
    Map<String, WebRouteInfo> webRouteMap = SpringBootstrap.getInstance().getRpcApp().getBootstrap().getWebRouteMap();
    WebRouteInfo webRouteInfo = webRouteMap.get(uri);
    if (Objects.nonNull(webRouteInfo) && !CollectionUtils.isEmpty(httpPrePluginList)) {
      for (HttpPrePlugin httpPrePlugin : httpPrePluginList) {
        retCode = httpPrePlugin.read(webRouteInfo, req, parmMap);
        if (retCode < 0) {
          break;
        }
      }
    }
    if (retCode == 0) {
      if (Objects.nonNull(webRouteInfo)) {
        TraceContext traceContext = Trace.newTraceContext();
        FlowReq flowReq = new FlowReq(parmMap);
        webRouteInfo.invoke(flowReq).thenAccept(res -> {
          writeRes(ctx, req.uri(), res);
        });
        return;
      } else {
        flowRes.fail(-666).msg("路径未配置");
      }
    } else {
      flowRes.fail(retCode);
    }
    writeRes(ctx, req.uri(), flowRes);
  }

  private void writeRes(ChannelHandlerContext ctx, String url, FlowRes flowRes) {
    // 创建http响应
    FullHttpResponse response = new DefaultFullHttpResponse(
        HttpVersion.HTTP_1_1,
        HttpResponseStatus.OK);

    // 设置头信息
    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=UTF-8")
        .set("server", "rpc-http-server 0.0.2").set("date", System.currentTimeMillis());
    response.headers().set("Access-Control-Allow-Origin", "*");
    response.headers().set("Access-Control-Allow-Methods", "*");
    response.headers().set("Access-Control-Max-Age", "3600");
    response.headers().set("Access-Control-Allow-Headers", "*");
    response.headers().set("Access-Control-Allow-Credentials", "true");

    if (!CollectionUtils.isEmpty(httpRetPluginList)) {
      for (HttpRetPlugin r : httpRetPluginList) {
        r.write(response, flowRes.toMap());
      }
    }

    String msg = JsonUtil.toJson(flowRes.toMap());

    ctx.writeAndFlush(response.replace(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8)))
        .addListener(ChannelFutureListener.CLOSE);
  }
}
